package cgl.hadoopsensorgrid.sensorgrid.util;

import java.io.IOException;
import java.io.DataInput;
import java.io.DataOutput;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

/***
 * 
 * This class store bytes and its length, along with an assigned ID. This class
 * is designed to be used as keys of data pairs.
 * 
 * @author Chao Sun
 * 
 */
public class serializedText implements WritableComparable<serializedText> {
	private int partitionID;
	private byte[] bytes;
	private int ID;
	private int length;

	private byte[] secondaryBytes;
	private int secondaryLength;

	/**
	 * Constructor.
	 */
	public serializedText() {
	}

	/**
	 * Constructor.
	 */
	public serializedText(int pId, byte[] by, int Id) {
		partitionID = pId;
		if (by != null) {
			length = by.length;
			bytes = new byte[length];
			System.arraycopy(by, 0, bytes, 0, length);
		}
		ID = Id;
	}

	/**
	 * Constructor.
	 */
	public serializedText(int pId, byte[] by, int Id, byte[] secBy) {
		partitionID = pId;
		if (by != null) {
			length = by.length;
			bytes = new byte[length];
			System.arraycopy(by, 0, bytes, 0, length);
		}
		ID = Id;

		if (secBy != null) {
			secondaryLength = secBy.length;
			secondaryBytes = new byte[secondaryLength];
			System.arraycopy(secBy, 0, secondaryBytes, 0, secondaryLength);
		}
	}

	/**
	 * Returns the partitionID.
	 */
	public int getPID() {
		return partitionID;
	}

	/**
	 * Returns the raw bytes.
	 */
	public byte[] getBytes() {
		return bytes;
	}

	/**
	 * Returns the raw bytes as String.
	 */
	public String toString() {
		return new String(bytes);
	}

	/**
	 * Returns the ID.
	 */
	public int getID() {
		return ID;
	}

	/**
	 * Returns the number of bytes in the byte array.
	 */
	public int getLength() {
		return length;
	}
	
	/**
	 * Returns the secondary byte array.
	 */
	public byte[] getSecBytes() {
		return secondaryBytes;
	}
	
	/**
	 * Returns the secondary byte array as String.
	 */
	public String secToString() {
		return new String(secondaryBytes);
	}
	
	/**
	 * Returns the number of bytes in the secondary byte array.
	 */
	public int getSecLength() {
		return secondaryLength;
	}

	/**
	 * Set to a by byte array
	 * 
	 * @param by
	 *          the data to copy from
	 * @param Id
	 *          the ID of this data
	 */
	public void set(int pId, byte[] by, int Id) {
		partitionID = pId;
		length = by.length;
		bytes = new byte[length];
		System.arraycopy(by, 0, bytes, 0, length);
		ID = Id;
	}

	/**
	 * Set to a by byte array
	 * 
	 * @param by
	 *          the data to copy from
	 * @param Id
	 *          the ID of this data
	 * @param secBy
	 *          the data for the secondary byte array
	 */
	public void set(int pId, byte[] by, int Id, byte[] secBy) {
		partitionID = pId;
		length = by.length;
		bytes = new byte[length];
		System.arraycopy(by, 0, bytes, 0, length);
		ID = Id;

		secondaryLength = secBy.length;
		secondaryBytes = new byte[secondaryLength];
		System.arraycopy(secBy, 0, secondaryBytes, 0, secondaryLength);
	}

	/**
	 * Read the fields.
	 * 
	 * @param in
	 *          the input data
	 */
	@Override
	public void readFields(DataInput in) throws IOException {
		partitionID = in.readInt();
		length = in.readInt();
		if (length != 0) {
			bytes = new byte[length];
			in.readFully(bytes, 0, length);
		}
		ID = in.readInt();
		secondaryLength = in.readInt();
		if (secondaryLength != 0) {
			secondaryBytes = new byte[secondaryLength];
			in.readFully(secondaryBytes, 0, secondaryLength);
		}
	}

	/**
	 * Write the fields.
	 * 
	 * @param out
	 *          the output data
	 */
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(partitionID);
		out.writeInt(length);
		if (length != 0) {
			out.write(bytes, 0, length);
		}
		out.writeInt(ID);
		out.writeInt(secondaryLength);
		if (secondaryLength != 0) {
			out.write(secondaryBytes, 0, secondaryLength);
		}
	}

	/**
	 * Provide a method for comparing two instances.
	 */
	@Override
	public int compareTo(serializedText other) {
		int thisPID = partitionID;
		int thatPID = other.getPID();
		if (thisPID < thatPID) {
			return -1;
		} else if (thisPID > thatPID) {
			return 1;
		}

		int thisValue = ID;
		int thatValue = other.getID();
		return (thisValue < thatValue ? -1 : (thisValue == thatValue ? 0 : 1));
	}

	/***
	 * 
	 * A Comparator that compares serialized IntPair.
	 * 
	 * @author Chao Sun
	 * 
	 */
	public static class Comparator extends WritableComparator {

		public Comparator() {
			super(serializedText.class);
		}

		public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
			return compareBytes(b1, s1, l1, b2, s2, l2);
		}

	}

	// register this comparator
	static {
		WritableComparator.define(serializedText.class, new Comparator());
	}

	/**
	 * Check if two objects are equal to each other.
	 * 
	 * @param obj
	 *          the object to be compared with this instance.
	 */
	public boolean equals(Object obj) {
		if (obj instanceof serializedText
				&& partitionID == ((serializedText) obj).getPID()) {
			String selfBytes = new String(bytes);
			String otherBytes = new String(((serializedText) obj).getBytes());
			String selfSecBytes = new String(secondaryBytes);
			String otherSecBytes = new String(((serializedText) obj).getSecBytes());
			if (selfBytes.equals(otherBytes) && ID == ((serializedText) obj).getID() &&
					selfSecBytes.equals(otherSecBytes)) {
				return true;
			} else {
				return false;
			}
		} else {
			return false;
		}
	}

	/**
	 * Check if the instance has been initialized.
	 */
	public boolean isInitialized() {
		if (partitionID != 0 && bytes != null && ID != 0 && length == bytes.length) {
			return true;
		} else {
			return false;
		}
	}

	/**
	 * Check if the instance has a secondary byte array.
	 */
	public boolean hasSecondaryBtyes() {
		if (secondaryBytes != null && secondaryLength == secondaryBytes.length) {
			return true;
		} else {
			return false;
		}
	}

}